package com.ctsi.flume.interceptor;

import static com.ctsi.flume.interceptor.FilenameInterceptor.Constants.FILENAME;
import static com.ctsi.flume.interceptor.FilenameInterceptor.Constants.PRESERVE;
import static com.ctsi.flume.interceptor.FilenameInterceptor.Constants.PRESERVE_DFLT;



import java.util.List;
import java.util.Map;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;
import org.apache.flume.interceptor.TimestampInterceptor;

public class FilenameInterceptor implements Interceptor {

	  private final boolean preserveExisting;

	  /**
	   * Only {@link FilenameInterceptor.Builder} can build me
	   */
	  private FilenameInterceptor(boolean preserveExisting) {
	    this.preserveExisting = preserveExisting;
	  }

	  @Override
	  public void initialize() {
	    // no-op
	  }

	  /**
	   * Modifies events in-place.
	   */
	  @Override
	  public Event intercept(Event event) {
		//System.out.print("Intercept...");
	    Map<String, String> headers = event.getHeaders();
	    if (headers.containsKey(Constants.FILENAME)) {
	    	String filename=headers.get(Constants.FILENAME);
	    	
	    	String ts=filename.split("_")[2];
	    	String date=ts.substring(0,8);
	    	String hour=ts.substring(8,10);
	    	String interval=ts.substring(10, 12);
	    	 
	    	headers.put("date", date);
	    	headers.put("hour", hour);
	    	headers.put("interval", interval);
	    	 
	    } else {
	      
	      
	    }
	    return event;
	  }

	  /**
	   * Delegates to {@link #intercept(Event)} in a loop.
	   * @param events
	   * @return
	   */
	  @Override
	  public List<Event> intercept(List<Event> events) {
	    for (Event event : events) {
	      intercept(event);
	    }
	    return events;
	  }

	  @Override
	  public void close() {
	    // no-op
	  }

	  /**
	   * Builder which builds new instances of the TimestampInterceptor.
	   */
	  public static class Builder implements Interceptor.Builder {

	    private boolean preserveExisting = Constants.PRESERVE_DFLT;

	    @Override
	    public Interceptor build() {
	      return new FilenameInterceptor(preserveExisting);
	    }

	    @Override
	    public void configure(Context context) {
	      preserveExisting = context.getBoolean(Constants.PRESERVE, Constants.PRESERVE_DFLT);
	    }

	  }

	  public static class Constants {
	    public static String FILENAME = "basename";
	    public static String PRESERVE = "preserveExisting";
	    public static boolean PRESERVE_DFLT = false;
	  }
}
